import time
from urllib.error import HTTPError

from openmind_hub import create_repo, delete_repo, repo_info, dataset_info, get_full_repo_name, \
    metadata_update
import unittest
from openmind_hub import RepositoryNotFoundError, GatedRepoError
from openmind_hub.plugins.openmind.utils._error import BadRequestError
from data import license_data
from ddt import ddt, data


token = '123456789()'
repo_id = "guoxiaozhen2/dataset_test2024"
license = "mit"

@ddt
class TestOpenmindHub(unittest.TestCase):
    def test_01_create_datasetrepo_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            a = repo_info(
                repo_id=repo_id,
                token=token,
                revision="main",
                repo_type="dataset",
                timeout=None,
                files_metadata=False
            )
            self.assertIsNotNone(a, "仓库不存在，创建失败")
        except:
            raise Exception("创建仓库有报错")
        else:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_02_create_datasetrepo_duplicate_existok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
        except:
            raise Exception("exist_ok=True未生效")
        else:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_03_create_datasetrepo_duplicate_existnotok(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
            raise Exception("exist_ok=False未生效")
        except BadRequestError:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_04_delete_datasetrepo_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
            time.sleep(3)
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
            repo_info(
                repo_id=repo_id,
                token=token,
                revision="main",
                repo_type="dataset",
                timeout=None,
                files_metadata=False
            )
            raise Exception("删除仓库有报错")
        except RepositoryNotFoundError:
            pass
    def test_05_delete_mdatasetrepo_duplicate_missok(self):
        try:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
        except:
            raise Exception("missing_ok=True未生效")
    def test_06_delete_datasetrepo_duplicate_missnotok(self):
        try:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=False
            )
            raise Exception("missing_ok=False未生效")
        except RepositoryNotFoundError:
            pass
    def test_07_create_datasetrepo_wrong_token(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token+"a",
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
            raise Exception("token未校验")
        except GatedRepoError:
            pass
    def test_08_create_datasetrepo_wrong_license(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license+"s",
                token=token,
                repo_type="dataset",
                exist_ok=False,
                fullname=None
            )
            raise Exception("license未校验")
        except BadRequestError:
            pass
    def test_09_dataset_info_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            a=dataset_info(
                repo_id=repo_id,
                timeout=None,
                token=token,
            )
            self.assertIsNotNone(a,"仓库不存在，创建失败")
        except:
            raise Exception("获取数据库信息有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_10_dataset_info_noexist_ropo(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            dataset_info(
                repo_id=repo_id+"a",
                timeout=None,
                token=token,
            )
            raise Exception("repo未校验")
        except RepositoryNotFoundError:
            pass
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_11_dataset_info_wrong_token(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            dataset_info(
                repo_id=repo_id,
                timeout=None,
                token=token+"a",
            )
            raise Exception("token未校验")
        except GatedRepoError:
            pass
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_12_get_full_repo_name_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            a = get_full_repo_name(
                model_id=repo_id,
                organization=None,
                token=token,
            )
            self.assertEqual(a,repo_id,"仓库不存在，创建失败")
        except:
            raise Exception("获取数据库信息有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_13_get_full_repo_name_notoken(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            get_full_repo_name(
                model_id=repo_id,
                organization=None,
                token=None,
            )
        except HTTPError:
            pass
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )
    def test_14_metadata_update_success(self):
        try:
            create_repo(
                repo_id=repo_id,
                private=True,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            metadata_update(
                repo_id=repo_id,
                metadata={"tags": ["music", "legal"], "task_categories": ["text-retrieval"], "license": "mit",
                 "size_categories": "n<1K"},
                overwrite=True,
                repo_type="dataset",
                token=token,
            )
        except :
            raise Exception("更新仓库标签有报错")
        finally:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )

    @data(*license_data.values())
    def test_15_create_datasetrepo_morelicenses(self,data):
        license = data["license"]
        print(license)
        try:
            create_repo(
                repo_id=repo_id,
                private=False,
                desc="描述内容",
                license=license,
                token=token,
                repo_type="dataset",
                exist_ok=True,
                fullname=None
            )
            a = repo_info(
                repo_id=repo_id,
                token=token,
                revision="main",
                repo_type="dataset",
                timeout=None,
                files_metadata=False
            )
            self.assertIsNotNone(a, "仓库不存在，创建失败")
        except:
            raise Exception("创建仓库有报错")
        else:
            delete_repo(
                repo_id=repo_id,
                token=token,
                repo_type="dataset",
                missing_ok=True
            )

if __name__ == '__main__':
    unittest.main()